package com.bupt.clusterlistener.handler;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bupt.clusterlistener.config.ApiClientLoader;
import com.bupt.clusterlistener.config.ClusterIP;
import com.bupt.clusterlistener.constants.KafkaConsts;
import com.bupt.clusterlistener.message.PlaceResult;
import io.kubernetes.client.custom.IntOrString;
import io.kubernetes.client.custom.Quantity;
import io.kubernetes.client.openapi.ApiClient;
import io.kubernetes.client.openapi.ApiException;
import io.kubernetes.client.openapi.apis.CoreV1Api;
import io.kubernetes.client.openapi.models.*;
import lombok.Data;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @author soonmachine
 */
@Component
@Slf4j
public class MessageHandler {


    @Autowired
    private ClusterIP clusterIP;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;


    @Autowired
    private ApiClientLoader apiClientLoader;

    private static final Map<String, ApiClientLoader.KubeConfigPath> API_CLIENT_MAP = new HashMap<>();

    static {
        API_CLIENT_MAP.put("k8snode1", ApiClientLoader.KubeConfigPath.NODE1);
        API_CLIENT_MAP.put("k8snode2", ApiClientLoader.KubeConfigPath.NODE2);
        API_CLIENT_MAP.put("k8snode3", ApiClientLoader.KubeConfigPath.CLOUD);
    }



    private static final Map<String, CRDInfo> CRD_MAP = new HashMap<>();


    @KafkaListener(topics = KafkaConsts.NODE_COMMAND, containerFactory = "ackContainerFactory")
    public void handleMessage(ConsumerRecord record, Acknowledgment acknowledgment){
        try {
            String message = (String) record.value();
            JSONObject jsonObject = JSON.parseObject(message);
            log.info("receive message:{}", jsonObject.toJSONString());
            String state = jsonObject.getString("state");
            switch (state){
                case "place":
                    placeTask(jsonObject);
                    break;
                case "remove":
                    removeTask(jsonObject, null);
                    break;
                case "upload":
                    uploadTask(jsonObject);
                    break;
                default:
                    break;
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            acknowledgment.acknowledge();
        }
    }

    private void uploadTask(JSONObject jsonObject) {
        String taskId = jsonObject.getString("taskId");
        CRDInfo info = CRD_MAP.get(taskId);
        if (info != null){
        ApiClient apiClient = apiClientLoader.getApiClientByPath(API_CLIENT_MAP.get(info.getNode()));
        removeTask(jsonObject, apiClient);
        }
        placeTask(jsonObject);
    }

    private void removeTask(JSONObject jsonObject, ApiClient client) {
        try {
            PlaceResult placeResult = new PlaceResult().setIsPid(jsonObject.getString("ispid")).setRequestId(jsonObject.getString("taskId")).setState("remove").setCpuSize(jsonObject.getDouble("cpuSize")).setMemorySize(jsonObject.getDouble("memorySize")).setTimeStamp(String.valueOf(System.currentTimeMillis()));
            kafkaTemplate.send(KafkaConsts.PLACE_RESULT, JSON.toJSONString(placeResult));
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        //get apiClient
        String node = jsonObject.getString("node");
        if (!API_CLIENT_MAP.containsKey(node)){
            log.error("放置决策消息异常，获取apiClient异常");
        }
        ApiClient apiClient = apiClientLoader.getApiClientByPath(API_CLIENT_MAP.get(node));
        apiClient.setDebugging(true);
        if (client != null){
            apiClient = client;
        }
        CoreV1Api apiInstance = new CoreV1Api(apiClient);

        //get taskId
        String taskId = jsonObject.getString("taskId");

        CRDInfo info = CRD_MAP.get(taskId);
        if (info == null){
            log.error("删除任务异常，获取crdInfo异常");
            return;
        }
        deletePod(apiInstance, info);
        deleteService(apiInstance, info);
        CRD_MAP.remove(taskId);
    }

    public void placeTask(JSONObject jsonObject){
        //get apiClient
        String node = jsonObject.getString("node");
        if (!API_CLIENT_MAP.containsKey(node)){
            log.error("放置决策消息异常，获取apiClient异常");
        }
        ApiClient apiClient = apiClientLoader.getApiClientByPath(API_CLIENT_MAP.get(node));
        apiClient.setDebugging(true);
        CoreV1Api apiInstance = new CoreV1Api(apiClient);


        V1Pod namespacedPod = createPod(apiInstance, jsonObject);
        if (namespacedPod == null){
            return;
        }
        String podName = jsonObject.getString("podName");
        String requestId = jsonObject.getString("taskId");
        CRDInfo crdInfo = new CRDInfo().setNode(node);
        crdInfo.setPodName(Objects.requireNonNull(namespacedPod.getMetadata()).getName());
        V1Service namespacedService = createService(apiInstance, jsonObject);
        if (namespacedService == null){
            deletePod(apiInstance, crdInfo);
            return;
        }
        crdInfo.setServiceName(Objects.requireNonNull(namespacedService.getMetadata()).getName());
        CRD_MAP.put(requestId, crdInfo);
        String ip = clusterIP.getIp(node);

        //send message
        List<V1ServicePort> ports = Objects.requireNonNull(namespacedService.getSpec()).getPorts();
        assert ports != null;
        V1ServicePort v1ServicePort = ports.get(0);
        Integer port1 = v1ServicePort.getNodePort();
        String url = ip + ":" +port1;
        String time = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
        String isPid = jsonObject.getString("ispid");
        PlaceResult placeResult = new PlaceResult().setNode(node).setIsPid(isPid).setRequestId(requestId).setState("place").setCpuSize(jsonObject.getDouble("cpuSize")).setMemorySize(jsonObject.getDouble("memorySize")).setUrl(url).setTimeStamp(String.valueOf(System.currentTimeMillis()));
        kafkaTemplate.send(KafkaConsts.PLACE_RESULT, JSON.toJSONString(placeResult));
    }

    public V1Pod createPod(CoreV1Api apiInstance, JSONObject jsonObject){
        //set pod port
        int containerPort = jsonObject.getInteger("containerPort");
        List<V1ContainerPort> port = new ArrayList<>();
        port.add(new V1ContainerPort().containerPort(containerPort));

        String image = jsonObject.getString("image");

        //set pod resource
        Double cpu = jsonObject.getDouble("cpuSize");
        String cpuLimit = cpu.intValue() + "m";
        Double memory = jsonObject.getDouble("memorySize");
        String memoryLimit = memory.intValue() + "Mi";
        Map<String, Quantity> limits = new HashMap<>();
        limits.put("cpu", Quantity.fromString(cpuLimit));
        limits.put("memory", Quantity.fromString(memoryLimit));
        V1ResourceRequirements resourceRequirements = new V1ResourceRequirements();
        resourceRequirements.setLimits(limits);

        String podName = jsonObject.getString("podName");
        Map<String, String> nodeLabel = new HashMap<String, String>() {{
            put("name", podName);
        }};


        V1Pod v1Pod = new V1Pod();
        v1Pod.setApiVersion("v1");
        v1Pod.setKind("Pod");
        V1ObjectMeta metadata = new V1ObjectMeta();
        metadata.setLabels(nodeLabel);
        metadata.setName(podName);
        v1Pod.setMetadata(metadata);
        V1PodSpec spec = new V1PodSpec();

        V1Container container = new V1Container();
        container.setName(podName);
        container.setImage(image);
        container.setImagePullPolicy("IfNotPresent");
        container.setResources(resourceRequirements);
        container.setPorts(port);
        spec.setContainers(Collections.singletonList(container));
        v1Pod.setSpec(spec);



        try {
            V1Pod namespacedPod = apiInstance.createNamespacedPod("app", v1Pod, null, null, null, null);
            if (namespacedPod == null){
                log.error("创建pod失败");
                return null;
            }
            log.info("创建pod成功, podName: " + podName);
            return namespacedPod;
        } catch (ApiException e) {
            log.error("Exception when calling CoreV1Api#createNamespacedPod {}", e.getMessage());
            return null;
        }
    }

    public V1Service createService(CoreV1Api apiInstance, JSONObject jsonObject){

        //set pod port
        int containerPort = jsonObject.getInteger("containerPort");


        String podName = jsonObject.getString("podName");
        Map<String, String> nodeLabel = new HashMap<String, String>() {{
            put("name", podName);
        }};

        V1Service v1Service = new V1Service();

        v1Service.setApiVersion("v1");
        v1Service.setKind("Service");

        V1ObjectMeta metadata = new V1ObjectMeta();
        metadata.setLabels(nodeLabel);
        metadata.setName(podName);
        v1Service.setMetadata(metadata);

        V1ServiceSpec spec = new V1ServiceSpec();
        spec.setType("NodePort");
        spec.setSelector(Collections.singletonMap("name", podName));

        V1ServicePort servicePort = new V1ServicePort();
        servicePort.setPort(containerPort);
        servicePort.setTargetPort(new IntOrString(containerPort));
        servicePort.setProtocol("TCP");
        spec.setPorts(Collections.singletonList(servicePort));

        v1Service.setSpec(spec);

        try {
            V1Service namespacedService = apiInstance.createNamespacedService("app", v1Service, null, null, null, null);

            if (namespacedService == null){
                log.error("创建service失败");
                return null;
            }
            log.info("创建service成功, serviceName: " + podName);
            return namespacedService;
        } catch (ApiException e) {
            log.error("Exception when calling CoreV1Api#createNamespacedService");
            return null;
        }
    }

    public void deletePod(CoreV1Api apiInstance, CRDInfo crdInfo){
        try {
            apiInstance.deleteNamespacedPod(crdInfo.getPodName(), "app", null, null, null, null, null, null);
        } catch (Exception exception) {
            log.error("Exception when calling CoreV1Api#deleteNamespacedPod {}", exception.getMessage());
        }
    }

    public void deleteService(CoreV1Api apiInstance, CRDInfo crdInfo){
        try {
             apiInstance.deleteNamespacedService(crdInfo.getServiceName(), "app", null, null, null, null, null, null);
        } catch (Exception e) {
            log.error("Exception when calling CoreV1Api#deleteNamespacedService {}", e.getMessage());
        }
    }

    public void clearNamespace(){
        apiClientLoader.getApiClientMap().values().stream().forEach(
                apiClient -> {
                    CoreV1Api apiInstance = new CoreV1Api(apiClient);
                    try {
                        apiInstance.deleteNamespacedPod("app", null, null, null, null, null, null, null);
                        apiInstance.deleteNamespacedService("app", null, null, null, null, null, null, null);
                    } catch (ApiException e) {
                        log.error("Exception when calling CoreV1Api#deleteNamespacedPod {}", e.getMessage());
                    }
                }
        );
    }

    @Data
    @Accessors(chain = true)
    static class CRDInfo{
        private String node;
        private String podName;
        private String serviceName;
    }

}


